package edu.purdue.cs.aggr.accumulators;

import java.io.IOException;
import java.io.ObjectInput;
import java.io.ObjectOutput;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;

import edu.purdue.cs.aggr.AbstractAccumulator;
import edu.purdue.cs.aggr.Accumulator;

public class SortAccumulator extends AbstractAccumulator implements Accumulator<Long, LinkedList<Long>> {

	private LinkedList<Long> data = new LinkedList<Long>();
	private boolean sorted = false;
	
	public SortAccumulator(String sourceId)
  {
	  super(sourceId);
  }

  @Override
	public void writeExternal(ObjectOutput out) throws IOException {
    super.writeExternal(out);
		out.writeObject(data);
		out.writeBoolean(sorted);
	}

	@Override
	public void readExternal(ObjectInput in) throws IOException,
			ClassNotFoundException {
	  super.readExternal(in);
		data = (LinkedList<Long>) in.readObject();
		sorted = in.readBoolean();
	}

	@Override
	public void initialize(Long input) {
		data.add(input);
	}

	@Override
	public void addValue(Long input) {
		data.add(input);
		sorted = false;
	}

	@Override
	public LinkedList<Long> getAccumulation() {
	  if (!sorted) {
		sort();
		sorted = true;
      }
	  return data;
	}

	@Override
	public void addAccumulators(List<Accumulator<Long, LinkedList<Long>>> accumulators) {
		
		//TODO
		
//		LinkedList<Long> data1 = accumulators.getAccumulation();
//		
//		int totalSize = data.size() + data1.size();
//		long[] elementArr = new long[totalSize];
//		long[] supportArr = new long[totalSize];
//		
//		Iterator<Long> iter = data.iterator();
//		int i = 0;
//		while (iter.hasNext()) {
//			long l = iter.next();
//			elementArr[i] = l;
//			supportArr[i] = l;
//			i++;
//		}
//		
//		iter = data1.iterator();
//		while (iter.hasNext()) {
//			long l = iter.next();
//			elementArr[i] = l;
//			supportArr[i] = l;
//			i++;
//		}
//		
//		merge(elementArr, supportArr, 0, data.size() - 1, totalSize - 1);
	}
	
	private void sort() {
		long[] elementArr = new long[data.size()];
		long[] supportArr = new long[data.size()];
		
		int i = 0;
		Iterator<Long> iter = data.iterator();
		while (iter.hasNext()) {
			long l = iter.next();
			elementArr[i] = l;
			supportArr[i] = l;
			i++;
		}
		
		mergeSort(elementArr, supportArr, 0, elementArr.length - 1);
	}
	
    private static void mergeSort(long[] valueArr, long[] supportArr, int start, int end) {
        if (start == end) {
            return;
        }
        
        int middle = (start + end) / 2;
        mergeSort(valueArr, supportArr, start, middle);
        mergeSort(valueArr, supportArr, middle + 1, end); // Java always gives the floor int as middle.
        merge (valueArr, supportArr, start, middle, end);
    }

    private static void merge(long[] valueArr, long[] supportArr, int start, int middle, int end) {
        int leftTemp = start;
        int rightTemp = middle + 1;
        int temp = start;
        
        while (true) {
            if (leftTemp > middle && rightTemp > end) {
                break;
            } else if (leftTemp <=middle && rightTemp <=end) {
                if ( valueArr[leftTemp] <= valueArr[rightTemp]) {
                    supportArr[temp]  = valueArr[leftTemp];
                    leftTemp++;
                } else {
                    supportArr[temp] = valueArr[rightTemp];
                    rightTemp++;
                }
                temp++;
            } else if (rightTemp > end) {
                supportArr[temp]  = valueArr[leftTemp];
                leftTemp++;
                temp++;
            } else if (leftTemp > middle) {
                // No need to copy here since value is already in the correct place
                rightTemp++;
                temp++;
            }
        }
        
        for (int i = start; i <=end; i++) {
            valueArr[i] = supportArr[i];
        }
    }
    
    public static void main(String[] args) {
        // Testing merge sort
        long[] valueArr = {70, 80, 100, 80, 50, 40, 98, 103, 20, 10, 99, 74};
        long[] supportArr = new long[valueArr.length];
        System.arraycopy(valueArr, 0, supportArr, 0, valueArr.length);
        
//        TopKMapper.K = 5;
        mergeSort(valueArr, supportArr, 0, valueArr.length - 1);
        
        for (int i=0; i<valueArr.length; i++) {
        	System.out.print(valueArr[i] + ", ");
        }
    }

	public boolean isSorted() {
		return sorted;
	}

	public void setSorted(boolean sorted) {
		this.sorted = sorted;
	}

}
